package org.example;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j(topic = "c.TestPool")
public class TestPool {
    public static void main(String[] args) {
        ThreadPool threadPool = new ThreadPool(2, 1000, TimeUnit.MILLISECONDS , 10);
        for (int i = 0; i < 15; i++) {
            int j = i;
            threadPool.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        Thread.sleep(1000000000);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                    log.debug("{}", j);
                }
            });
        }
    }

}
@Slf4j(topic = "c.ThreadPool")
class ThreadPool{

    //任务队列
    private BlockingQueue<Runnable> taskQueue;

    //线程集合
    private HashSet<Worker> workers = new HashSet<>();

    //核心线程数
    private int coreSize;

    //获取任务的超时时间
    private long timeout;

    private TimeUnit timeUnit;

    public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit, int capacity) {
        this.coreSize = coreSize;
        this.timeout = timeout;
        this.timeUnit = timeUnit;
        this.taskQueue = new BlockingQueue<>(capacity);
    }

    //执行任务
    public void execute(Runnable task){
        //任务数没有超过coreSize时, 交给Worker对象执行
        //超过coreSize时, 加入任务队列暂存
        synchronized (workers) {
            if(workers.size() < coreSize){
                Worker worker = new Worker(task);
                log.debug("新增 worder: {}, {}", worker, task);
                workers.add(worker);
                worker.start();
            }else{
                taskQueue.put(task);
            }
        }
    }

    class Worker extends Thread{
        private Runnable task;

        public Worker(Runnable task){
            this.task = task;
        }

        @Override
        public void run() {
            //执行任务
            //1.任务不为空,执行任务
            //2.当任务执行完毕, 再接着从任务队列获取任务并执行
            //while(task != null || (task = taskQueue.take()) != null){
            while(task != null || (task = taskQueue.poll(timeout,timeUnit)) != null){
                log.debug("正在执行...{}",task);
                try {
                    task.run();
                }catch (Exception e){
                    e.printStackTrace();
                } finally {
                    task = null;
                }
            }
            synchronized (workers){
                log.debug("worker被移除{}", this);
                workers.remove(this);
            }
        }
    }

}
@Slf4j(topic = "c.BlockingQueue")
class BlockingQueue<T>{
    //1.任务队列
    private Deque<T> queue = new ArrayDeque<>();

    //2.锁
    private ReentrantLock lock = new ReentrantLock();


    //3.生产者条件变量
    private Condition fullWaitSet = lock.newCondition();

    //4.消费者条件变量
    private Condition emptyWaitSet = lock.newCondition();

    //5.容量
    private int capacity;

    public BlockingQueue(int capacity) {
        this.capacity = capacity;
    }

    //带超时的阻塞获取
    public T poll(long timeout, TimeUnit unit){
        lock.lock();
        try{
            //转换为纳秒
            long nanos = unit.toNanos(timeout);
            while(queue.isEmpty()){
                if(nanos <= 0) return null;
                try {
                    nanos = emptyWaitSet.awaitNanos(nanos); //等待并更新剩余时间
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            fullWaitSet.signal();
            return queue.removeFirst();
        }finally {
            lock.unlock();
        }
    }

    //阻塞获取
    public T take(){
        lock.lock();
        try{
            while(queue.isEmpty()){
                try {
                    emptyWaitSet.await();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            fullWaitSet.signal();
            return queue.removeFirst();
        }finally {
            lock.unlock();
        }
    }

    //阻塞添加
    public void put(T element){
        lock.lock();
        try {
            while(queue.size() == capacity){
                try {
                    log.debug("等待加入任务队列... {}", element);
                    fullWaitSet.await();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            log.debug("加入任务队列 {}", element);
            queue.addLast(element);
            emptyWaitSet.signal();
        }finally {
            lock.unlock();
        }
    }

    //获取阻塞队列的大小
    public int size(){
        lock.lock();
        try{
            return queue.size();
        }finally {
            lock.unlock();
        }
    }
}
